package com.yytek.pm.tddb.task;

import com.yytek.common.client.base.MqMessage;
import com.yytek.common.core.constant.CacheConstants;
import com.yytek.common.core.utils.SpringUtils;
import com.yytek.common.redis.service.RedisService;

import com.yytek.pm.tddb.service.IotTddbHandleServiceImpl;
import lombok.extern.slf4j.Slf4j;


/**
 * 写入时序数据库任务线程
 */
@Slf4j
public class WriteTddbTask implements Runnable {

    //标识是否继续处理
    private boolean active = true;

    @Override
    public void run() {
        log.info("WriteTddbTask----:started");
        MqMessage line = null; //从队列中获取到的数据
        //获取处理时序数据库业务类
        IotTddbHandleServiceImpl tddbHandleService = SpringUtils.getBean(IotTddbHandleServiceImpl.class);
        //获取处理时序数据库业务类
        RedisService redisService = SpringUtils.getBean(RedisService.class);
        try {
            while (active) {
                    line = (MqMessage) redisService
                            .getQueueBrpop(CacheConstants.TDDB_INSERT_QUEUE_KEY,1);
                    if (line != null) {
                        //如果消息队列不为空,判断处理队列数据
                        tddbHandleService.processTddbLine(line);
                    } else if (tddbHandleService.hasTdQueueRedisValues()) {
                        //如果数据队列有值,批量写入数据
                        tddbHandleService.flushQueue();
                    } else {
                        // sleep a while to avoid high CPU usage if no more data in the queue and no buffered records, .
                        Thread.sleep(100);
                    }
            }
            if (tddbHandleService.hasTdQueueRedisValues()) {
                tddbHandleService.flushQueue();
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.info("line=%s 读取td队列报错",line);
            if (line != null) {
                redisService.setCacheQueue(CacheConstants.TDDB_INSERT_QUEUE_KEY,line);
            }
        }
    }

    public void stop() {
        this.active = false;
    }
}
